<?php

declare(strict_types=1);
/**
 * This file is part of MoChat.
 * @link     https://mo.chat
 * @document https://mochat.wiki
 * @contact  group@mo.chat
 * @license  https://github.com/mochat-cloud/mochat/blob/master/LICENSE
 */
namespace MoChat\App\WorkMessage\Queue;

use Hyperf\AsyncQueue\Annotation\AsyncQueueMessage;
use Hyperf\Di\Annotation\Inject;
use Hyperf\Logger\LoggerFactory;
use Hyperf\Redis\Redis;
use Hyperf\Utils\InteractsWithTime;
use MoChat\App\WorkMessage\Fetcher\MessageDataFetcher;
use MoChat\App\WorkMessage\Utils\MessageArchiveFactory;

class MessagePullQueue
{
    use InteractsWithTime;

    /**
     * @var \Psr\Log\LoggerInterface
     */
    protected $logger;

    /**
     * @var MessageArchiveFactory
     */
    protected $messageArchiveFactory;

    /**
     * @Inject
     * @var Redis
     */
    private $redis;

    public function __construct(LoggerFactory $loggerFactory)
    {
        $this->logger = $loggerFactory->get(__CLASS__);
    }

    /**
     * @AsyncQueueMessage(pool="message_pull")
     * @param int $corpId 企业ID
     */
    public function handle(int $corpId): void
    {
        try {
            $this->logger->debug(sprintf('拉取会话消息队列开始执行，企业ID: %s', (string) $corpId));
            if ($this->lock($corpId)) {
                $messageArchiveFactory = make(MessageArchiveFactory::class);
                $messageSdk = $messageArchiveFactory->make($corpId);
                $messageDataFetcher = make(MessageDataFetcher::class, [(int) $corpId]);
                $tmpData = $messageDataFetcher->pull(200, $messageSdk);
                if (! empty($tmpData)) {
                    $messageDataFetcher->syncData($tmpData);
                }
                $this->unlock($corpId);
            } else {
                $this->logger->debug(sprintf('中断，拉取会话消息队列有其他进程正在执行，企业ID: %s', (string) $corpId));
                return;
            }
        } catch (\Throwable $e) {
            $this->logger->error(sprintf('拉取会话消息队列失败，企业ID: %s，原因：%s', (string) $corpId, $e->getMessage()));
            $this->logger->error($e->getTraceAsString());
            $this->unlock($corpId);
        }
    }

    private function lock(int $corpId)
    {
        $seconds = $this->secondsUntil(60);
        $result = serialize(1);
        $option = ['nx'];
        if ($seconds > 0) {
            $option = [
                'nx',
                'ex' => $seconds,
            ];
        }

        return $this->redis->set($this->getLockKey($corpId), $result, $option);
    }

    private function getLock(int $corpId)
    {
        $result = $this->redis->get($this->getLockKey($corpId));
        if ($result === false) {
            return null;
        }
        return unserialize($result);
    }

    private function unlock(int $corpId)
    {
        $this->redis->del($this->getLockKey($corpId));
    }

    private function getLockKey(int $corpId)
    {
        return sprintf('message_pull_lock:%s', (string) $corpId);
    }
}
